package com.smsc.headend.task.engine.service.task.impl;


import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.extension.conditions.update.LambdaUpdateChainWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.smsc.headend.module.common.entity.DeviceOperation;
import com.smsc.headend.module.data.enums.DictionaryType;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.SubTaskGroup;
import com.smsc.headend.module.task.dto.SubTaskStatusDto;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.entity.DeviceTaskSub;
import com.smsc.headend.module.task.enums.SubTaskStatus;
import com.smsc.headend.task.engine.mapper.DeviceTaskSubMapper;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.task.DeviceTaskSubService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.nio.charset.Charset;
import java.util.List;

@Service
@Slf4j
public class DeviceTaskSubServiceImpl extends ServiceImpl<DeviceTaskSubMapper, DeviceTaskSub> implements DeviceTaskSubService {
    @Autowired
    AssetService assetService;
    @Autowired
    KafkaTemplate kafkaTemplate;

    @Override
    @Transactional
    public void batchAddDeviceSubTask(List<DeviceTaskSub> dtoList) {
        this.saveBatch(dtoList);
    }

    @Override
    public Integer getAtomicTaskGroupSeq(Task task, String atomicTaskNo) {
        Integer seq = 0;
        for (SubTaskGroup subTaskGroup : task.getSubTaskGroups()) {
            if (subTaskGroup.getSubTasks().stream().filter(at -> at.getAtomicTaskNo().equalsIgnoreCase(atomicTaskNo)).findAny().isPresent()) {
                break;
            }
            seq++;
        }
        return seq;
    }

//    private DeviceTaskSub updateDeviceTaskSub(DeviceTaskSub deviceTaskSub) {
//        if (deviceTaskSub.getStatus().equals(SubTaskStatus.Processing.toString())) {
//            deviceTaskSub.setTvStart(System.currentTimeMillis() / 1000);
//        } else if (deviceTaskSub.getStatus().equals(SubTaskStatus.Success.toString()) || deviceTaskSub.getStatus().equals(SubTaskStatus.Failed.toString())) {
//            deviceTaskSub.setTvEnd(System.currentTimeMillis() / 1000);
//        }
//        deviceTaskSub.setTvCreate(null);
//        saveOrUpdate(deviceTaskSub);
//        return deviceTaskSub;
//    }


    @Override
    public void updateDeviceTaskSubStateAll(String taskNo, String status, Long timestamp) {
        lambdaUpdate().eq(DeviceTaskSub::getTaskNo, taskNo)
                .in(DeviceTaskSub::getStatus, SubTaskStatus.Processing.toString(), SubTaskStatus.Waiting.toString())
                .set(DeviceTaskSub::getStatus, status).set(DeviceTaskSub::getTvEnd, timestamp).update();
    }

    @Override
    public void updateSubDeviceTaskStatus(SubTaskStatusDto dto, String taskNo) {
//        DeviceTaskSub deviceTaskSub = lambdaQuery().eq(DeviceTaskSub::getDeviceTaskSeq, taskNo).eq(DeviceTaskSub::getSort, dto.getSort()).one();
//        if (deviceTaskSub == null) {
//            log.debug("no sub task {}", dto);
//            return;
//        }
        SubTaskStatus subTaskStatus = SubTaskStatus.forValue(dto.getStatus());
//        if (subTaskStatus == null) {
//            log.debug("no sub task status {}", dto);
//            return;
//        }
//        if (deviceTaskSub.getStatus().equalsIgnoreCase(subTaskStatus.toString())) {
//            return;
//        }
        LambdaUpdateChainWrapper<DeviceTaskSub> updateChainWrapper = lambdaUpdate().eq(DeviceTaskSub::getTaskNo, taskNo).eq(DeviceTaskSub:: getSort, dto.getSort())
                .set(DeviceTaskSub::getStatus, subTaskStatus.toString());

        if (dto.getTimestamp() == null) {
            updateChainWrapper.update();
            return;
        }

        switch (subTaskStatus) {
            case Waiting:
                break;
            case Processing:
                updateChainWrapper.set(DeviceTaskSub::getTvStart, dto.getTimestamp());
                break;
            default:
                updateChainWrapper.set(DeviceTaskSub::getTvEnd, dto.getTimestamp());
        }
        updateChainWrapper.update();
    }


    @Override
    public void addTaskSub(String taskNo, SubTaskGroup subTaskGroup, Long createTimeSecond, int sequence) {
        DeviceOperation dictionary = assetService.getDictionary(subTaskGroup.getGroupId(), DictionaryType.OperationGroup.getType());
        DeviceTaskSub deviceTaskSub = new DeviceTaskSub();
        deviceTaskSub.setSubTaskName(dictionary.getOptName());
//        deviceTaskSub.setDeviceTaskSeq(deviceTaskId);
        deviceTaskSub.setSort(sequence);
        deviceTaskSub.setTaskNo(taskNo);
        deviceTaskSub.setTvCreate(createTimeSecond);
        deviceTaskSub.setStatus(SubTaskStatus.Waiting.toString());
//        save(deviceTaskSub);
        kafkaTemplate.send(TaskKafkaTopic.SUB_TASK_CREATE, JSONUtil.toJsonStr(deviceTaskSub).getBytes(Charset.forName("utf-8")));
    }
}
